package flink.mysql_connect;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
public class JdbcSourceExample {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        //需要与获取的字段一一对应，否则会取不到值
        TypeInformation[] fieldTypes = new TypeInformation[] {
                BasicTypeInfo.LONG_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.INT_TYPE_INFO,
        };
        RowTypeInfo rowTypeInfo = new RowTypeInfo(fieldTypes);

        DataSet<Row> dataSource = env.createInput(JdbcInputFormat.buildJdbcInputFormat()
                .setDrivername("com.mysql.cj.jdbc.Driver")
                .setDBUrl("jdbc:mysql://127.0.0.1:3306/flink")
                .setUsername("root")
                .setPassword("root")
                .setQuery("select * from books")
                .setRowTypeInfo(rowTypeInfo)
                .finish());
        dataSource.print();
//        env.execute();
    }
}
